package utils;

import com.sun.org.apache.bcel.internal.generic.RET;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

import java.io.IOException;

public class MyKafkaUtil {
    private static final String BOOTSTRAP_SERVER = "39.98.72.89:9092";
    public static String getTopicDbDDL(String groupId){
        return "create table topic_db(\n" +
                "`database` string,\n" +
                "`table` string,\n" +
                "`type` string,\n" +
                "`ts` string,\n" +
                "`data` map<string,string>,\n" +
                "`old` map<string,string>\n" +
                ")"+getKafkaSourceConnOption("topic_db",groupId);

    }

    public static String getTopicDbDDLWithRT(String groupId,String timeType){
        return "create table topic_db(\n" +
                "`database` string,\n" +
                "`table` string,\n" +
                "`type` string,\n" +
                "`ts` string,\n" +
                "`data` map<string,string>,\n" +
                "`old` map<string,string>,\n" +
                "rt as to_timestamp("+timeType+"),\n" +
                "watermark for rt as rt - interval '2' second)"+getKafkaSourceConnOption("topic_db",groupId);

    }
    public static String getKafkaSourceConnOption(String topic,String groupId){
        return "with(\n" +
                "'connector' ='kafka',\n" +
                "'topic'='"+topic+"',\n" +
                "'properties.bootstrap.servers'='"+BOOTSTRAP_SERVER+"',\n" +
                "'properties.group.id'='"+groupId+"',\n" +
                "'scan.startup.mode'='latest-offset',\n" +
                "'format'='json'\n" +
                ")";

    }
    public static String getKafkaSinkConnOption(String topic){
        return "with(\n" +
                "'connector'='kafka',\n" +
                "'topic'='"+topic+"',\n" +
                "'properties.bootstrap.servers'='"+BOOTSTRAP_SERVER+"',\n" +
                "'format'='json')";
    }
    public static KafkaSource<String> getKafkaSource(String topic,String groupId){
        return KafkaSource.<String>builder()
                .setBootstrapServers(BOOTSTRAP_SERVER)
                .setTopics(topic)
                .setGroupId(groupId)
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new DeserializationSchema<String>() {
                    @Override
                    public TypeInformation<String> getProducedType() {
                        return BasicTypeInfo.STRING_TYPE_INFO;
                    }

                    @Override
                    public String deserialize(byte[] bytes) throws IOException {
                        if (bytes != null){
                            return new String(bytes);
                        }else {
                            return null;
                        }

                    }

                    @Override
                    public boolean isEndOfStream(String s) {
                        return false;
                    }
                })
                .build();
    }
    public static FlinkKafkaProducer<String> getFlinkKafkaProducer(String topic){
        return new FlinkKafkaProducer<String>(BOOTSTRAP_SERVER,topic,new SimpleStringSchema());
    }


}
